package org.sabayframework.cluster.impl.kryo;

import org.sabayframework.activity.Activity;
import org.sabayframework.activity.ActivityContext;
import org.sabayframework.activity.impl.Emit;
import org.sabayframework.activity.impl.StatefulActivity;
import org.sabayframework.channel.Channel;
import org.sabayframework.channel.Lock;
import org.sabayframework.cluster.DistributedActivity;
import org.sabayframework.log.Log;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;

/**
 * Asynchronously processes remote request
 * */
public final class RequestActivity extends StatefulActivity<Request,Request>{

	private final RequestManager requestManager;
	private final Kryo kryo;
	private Request request;

	public RequestActivity(RequestManager rm, Kryo kryo) {
		this.requestManager = rm;
		this.kryo = kryo;
	}
	

	@Override
	public int getOutputSize(ActivityContext<Request, Request> ctx) {
		return 0;
	}


	@Override
	public void input(ActivityContext<Request, Request> ctx, Lock<Request> inputLock) throws Exception {
		this.request = inputLock.get(0);
	}


	@SuppressWarnings("unchecked")
	@Override
	public void process(ActivityContext<Request, Request> ctx) throws Exception {
		Throwable error = null;
		Log.debug("Processing remote request", request);
		final DistributedActivityContext distributedCtx = request.getDistributedActivityContext();
		try{
			Object object = null;
			if(request.getData()!=null){
				Input in = new Input((byte[])request.getData());
				object = kryo.readClassAndObject(in);
			}
			switch (request.getType()) {
			case START:
				if(object!=null){
					Activity<Object, Object> a = (Activity<Object, Object>)object;
					DistributedActivity<Object, Object> da = distributedCtx.getDistributedActivity();
					da.setActivity(a);
					Channel<Object> input = (Channel<Object>) distributedCtx.getInput();
					Channel<Object> output =(Channel<Object>) distributedCtx.getOutput();
					ctx.getExecutionQueue().execute(a, input, output);
				}
				else error = new IllegalArgumentException("Activity must be provided");
				break;
			case CLOSE_INPUT:
				distributedCtx.setAlive(false);
				if(object!=null) {
					Throwable cause = (Throwable)object;
					distributedCtx.getInput().fail(cause);
				}else{
					distributedCtx.getInput().close();
				}
				break;
			case CLOSE_OUTPUT:
				if(object!=null) {
					Throwable cause = (Throwable)object;
					distributedCtx.getOutput().fail(cause);
				}else{
					distributedCtx.getOutput().close();
				}
				break;
			case INPUT:
				if(object!=null){
					Emit<Object> e = new Emit<Object>(false, object);
					Channel<Object> ch =  (Channel<Object>) distributedCtx.getInput();
					ctx.getExecutionQueue().execute(e, ch, ch);
				}
				break;
			case OUTPUT:
				if(object!=null){
					distributedCtx.endRequest();
					Emit<Object> e = new Emit<Object>(false,object);
					Channel<Object> ch =  (Channel<Object>) distributedCtx.getOutput();
					ctx.getExecutionQueue().execute(e, ch, ch);
				}
				break;
			default:
				error = new IllegalArgumentException("Uknown request type:" + request.getType());
				break;
			}
		} catch (Throwable e) {
			error = e;
		}
		//handle errors and send them back
		if(error != null){
			Log.error("Processing remote request", error);
			Request r = new Request(RequestType.CLOSE_OUTPUT, distributedCtx, error);
			requestManager.submitRequest(r);
		}
		request = null;
	}


	@Override
	public void output(ActivityContext<Request, Request> ctx, Lock<Request> outputLock) throws Exception {}

}